/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.protocol;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Properties;

import cgl.narada.discovery.broker.BrokerDiscoveryRequestForwardingService;
import cgl.narada.discovery.broker.BrokerDiscoveryRequestResponsePolicy;
import cgl.narada.discovery.broker.DefaultBrokerDiscoveryRequestResponsePolicy;
import cgl.narada.discovery.broker.messages.BrokerAdvertisement;
import cgl.narada.discovery.broker.messages.BrokerDiscoveryRequest;
import cgl.narada.discovery.broker.messages.BrokerDiscoveryResponse;
import cgl.narada.event.NBEvent;
import cgl.narada.node.BrokerNode;
import cgl.narada.node.BrokerSystemEventListener;
import cgl.narada.service.security.SecureTopicsManager;
import cgl.narada.service.security.kmc.SignedSecurityToken;
import cgl.narada.topicgenerator.LongTopicGenerator;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.TransportHandler;
import cgl.narada.transport.TransportHandlerImpl;
import cgl.narada.util.HashQueue;

/**
 * The ProtocolHandler provides the interface between the transport layer and
 * the protocol layer. It is also designed in a way to provide an interface
 * between the protocol layer and the application layer. This allows us to
 * decouple the transport aspects from the protocol layer. The protocol layer
 * thus deals only with the logical addresses associated with nodes, instead of
 * the various transport indentifiers (TCP,UDP <hostName, port> Multicast
 * <groupName, port>) associated with any given node.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 * 
 */

public class ProtocolHandler extends Thread implements ProtocolDebugFlags,
         Handler {

   /**
    * The queue of all messages received over various links across different
    * transport protocols
    */
   private LinkedList receivedEvents;

   private LinkedList receivedAudioEvents;

   private Object syncObject;

   private boolean waiting = true;

   /** An individual dataUnit within the queue of received messages */
   private DataReceived dataReceived;

   private short[] connectionVector;

   /** The nodeAddress of this node */
   private NodeAddress thisNodeAddress;

   /** Node Addition protocol */
   private NodeAdditionProtocol nodeAdditionProtocol;

   /** EventRouting protocol */
   private TagValueEventRoutingProtocol tvEventRoutingProtocol;

   /** RtpEventRouting protocol */
   private RtpEventRoutingProtocol rtpEventRoutingProtocol;

   private JmsProfilePropagation jmsProfilePropagation;

   /** GatewayPropagationProtocol */
   private GatewayPropagationProtocol gatewayPropagationProtocol;

   /** Connection Propagation */
   private ConnectionPropagation connectionPropagation;

   /** The gatewayInfo */
   private GatewayInfo gatewayInfo;

   /** Deals with the transport related aspects of communication */
   private TransportHandler transportHandler;

   /** The protocol ID factory */
   private ProtocolIDFactory protocolIDFactory;

   /** This destination */
   private Destinations thisDestination;

   /** Profile Propagation */
   private TagValueProfilePropagation tagvalueProfilePropagation;

   private RtpProfilePropagation rtpProfilePropagation;

   private XPathProfilePropagation xPathProfilePropagation;

   /** Client connection handler */
   private ClientConnectionHandler clientConnectionHandler;

   /** The Discovery request response policy verifier */
   private BrokerDiscoveryRequestResponsePolicy requestResponsePolicyHandler;

   /** A table of last known 1000 broker Discovery requests */
   private HashQueue discoveryRequestTable;

   /** Broker Discovery Request forwarding service */
   private BrokerDiscoveryRequestForwardingService fwdService;

   /** Plugin that manages a list of current secure topics, CRLs etc... */
   private SecureTopicsManager secureTopicsManager;

   // to process long topic generation events
   private LongTopicGenerator longTopicGenerator;

   int packetNumber = 0;

   private int brokerPort;

   private String discriminator;

   private String moduleName = "ProtocolHandler:";

   // HG:MGMT:
   /** Observer for link loss */
   private BrokerSystemEventListener brokerSystemEventListener;

   /**
    * JmsEventRouting protocol
    */
   private EventRoutingProtocol eventRoutingProtocol;

   private ProfilePropagationProtocol profilePropagationProtocol;

   private boolean nodeAddressAssigned = false;

   public boolean startForwardingService = true;

   private String configFileLocation = "";

   public ProtocolHandler() {
      syncObject = new Object();
      receivedEvents = new LinkedList();
      receivedAudioEvents = new LinkedList();
      transportHandler = new TransportHandlerImpl(this);

      // NOTE: This will just create the manager, However the init requires
      // reading the config file which is available in
      // setServiceConfigurationLocation() method, hence we do the init here
      secureTopicsManager = new SecureTopicsManager();

      // HG:MGMT
      brokerSystemEventListener = null;
   }

   /**
    * The protocol handler constructor.
    * 
    * @param nodeAddress
    *           The node address of the system.
    * @param connectionVector
    *           The number of active connections that are allowed at a time, at
    *           each level
    * @param concurrentConnectionLimit
    *           The limit on the number of active concurrent connections
    *           initiated by clients.
    * @param ipDiscriminator
    *           The discriminator used to determine if a node can be part of a
    *           cluster
    * @param gatewayInfo
    *           The connectivity graph of the hosting node.
    */
   public ProtocolHandler(NodeAddress nodeAddress, short[] connectionVector,
            int concurrentConnectionLimit, String ipDiscriminator,
            GatewayInfo gatewayInfo) {

      // HG:MGMT
      brokerSystemEventListener = null;

      thisNodeAddress = nodeAddress;
      this.gatewayInfo = gatewayInfo;
      this.connectionVector = connectionVector;
      discriminator = ipDiscriminator;
      nodeAdditionProtocol = new NodeAdditionProtocol(nodeAddress,
               connectionVector, ipDiscriminator, gatewayInfo);

      syncObject = new Object();
      receivedEvents = new LinkedList();
      receivedAudioEvents = new LinkedList();

      transportHandler = new TransportHandlerImpl(this);

      /**
       * The node should be open to handle connections only if it has been
       * assigned a destination_ID
       */
      clientConnectionHandler = new ClientConnectionHandler(this,
               concurrentConnectionLimit);

      longTopicGenerator = new LongTopicGenerator(thisNodeAddress,
               clientConnectionHandler);

      // HG: Security Manager !!
      secureTopicsManager = new SecureTopicsManager();
   }

   /**
    * After a node address has been assigned, this method is responsible for
    * setting the node address for the connectivity graph.
    * 
    * @param nodeAddress
    *           The node address of the server node hosting the connectivity
    *           graph
    */
   public void setAssignedAddress(NodeAddress nodeAddress) {
      thisNodeAddress = nodeAddress;
      nodeAddressAssigned = true;
      protocolIDFactory = new ProtocolIDFactory(thisNodeAddress, (short) 0);
      thisDestination = new Destinations(nodeAddress.getAddressInBytes());
      nodeAdditionProtocol.setNodeAddress(nodeAddress);
      gatewayInfo.setNodeAddress(thisNodeAddress);
      int[] addressAssgnInts = nodeAddress.getAddressInInts();
      int systemLevel = addressAssgnInts.length - 1;
      int[] addressConstruct = { addressAssgnInts[systemLevel - 0 /*
                                                                   * The level
                                                                   * of the node
                                                                   */] };

      gatewayInfo.setVertexNode(new Gateway(new NodeAddress(addressConstruct),
               (short) 0, true));
      connectionPropagation = new ConnectionPropagation(gatewayInfo, this);
      gatewayPropagationProtocol = new GatewayPropagationProtocol(
               connectionVector, gatewayInfo, this);

      /**
       * We need that this node be assigned an address, if this is not the case
       * propagation of subscription predicates and calculation of destinations
       * hierarchically is not possible
       */
      tagvalueProfilePropagation = new TagValueProfilePropagation(gatewayInfo,
               this);

      jmsProfilePropagation = new JmsProfilePropagation(gatewayInfo, this);

      profilePropagationProtocol = new ProfilePropagationProtocol(gatewayInfo,
               this);

      rtpProfilePropagation = new RtpProfilePropagation(gatewayInfo, this);

      xPathProfilePropagation = new XPathProfilePropagation(gatewayInfo, this);

      /**
       * This provides an indication that the node is now able to process client
       * connection requests
       */
      clientConnectionHandler.setNodeDestination(thisDestination);
      clientConnectionHandler.setVertexNode(gatewayInfo.getVertexNode());

      tvEventRoutingProtocol = new TagValueEventRoutingProtocol(nodeAddress,
               gatewayInfo, tagvalueProfilePropagation,
               clientConnectionHandler, this);

      rtpEventRoutingProtocol = new RtpEventRoutingProtocol(nodeAddress,
               gatewayInfo, rtpProfilePropagation, clientConnectionHandler,
               this);

      eventRoutingProtocol = new EventRoutingProtocol(nodeAddress, gatewayInfo,
               profilePropagationProtocol, clientConnectionHandler, this);

      longTopicGenerator.setNodeAddress(thisNodeAddress);
      
   }

   /**
    * shows whether node address is assigned
    */
   public boolean isNodeAddressAssigned() {
      return nodeAddressAssigned;
   }

   /**
    * Accessor function for retrieving the node address of the server node
    * hosting the connectivity graph.
    * 
    * @return The node address of the server node.
    */
   public NodeAddress getAssignedAddress() {
      return thisNodeAddress;
   }

   /**
    * Retreives the destination that has been assigned to this node. This is
    * simply a wrapper around the node address of the server node. This is very
    * useful in computing the destinations that have traversed by a message
    * packet.
    * 
    * @return The destinations correspinding to the server node.
    */
   public Destinations getNodeDestination() {
      return thisDestination;
   }

   /**
    * This method provides us with access to the ProtocolIDFactory which
    * provides us with a list of unique protocol ID's
    * 
    * @return The unique ID associated with a request/response. This is a unique
    *         ID system wide.
    */
   public ProtocolID getUniqueProtocolID() {
      if (protocolIDFactory == null) {
         System.out.println("ProtocolHandler: Attempt made to retrieve a "
                  + "protocolID on an uninitialized factory ");
         return null;
      }
      return protocolIDFactory.getProtocolID();
   }

   /** This designates this node as a gateway at level <i>level</i>. */
   public void setAsGatewayAtLevel(int level) {
      tagvalueProfilePropagation.setAsGatewayAtLevel(level);
      jmsProfilePropagation.setAsGatewayAtLevel(level);
      xPathProfilePropagation.setAsGatewayAtLevel(level);
      rtpProfilePropagation.setAsGatewayAtLevel(level);
      profilePropagationProtocol.setAsGatewayAtLevel(level);
   }

   public void unsetForwardingService() {
      startForwardingService = false;
   }

   /** This designates this node as a gateway at level <i>level</i>. */
   public void unsetAsGatewayAtLevel(int level) {
      System.out.println(moduleName + "Need to UNSET Gateways at level ("
               + level + ")");
   }

   /** This method indicates that the node in question is a broker */
   public boolean isClient() {
      return false;
   }

   private void processDataPackets(byte[] dataBytes, String sender) {
      int dataInfo = dataBytes[1];

      switch (dataInfo) {
         case (int) Protocol.NB_EVENT:
            // System.out.println(moduleName + "Dealing with event
            // propagations");
            // long startTime = System.currentTimeMillis();
            eventRoutingProtocol.processEventReceived(dataBytes);

         // long elapsedTime = System.currentTimeMillis() - startTime;
         // String report = moduleName + "Processing time for (" +
         // dataBytes.length + ") = " + elapsedTime;
         // System.out.println(report);
         break;

         case (int) Protocol.PROFILE_PROPAGATION:
            // System.out.println(moduleName + "Dealing with profile
            // propagations");

            // HG: Changed method signature to include the sender who sent this
            // request
            // We can thus notify the sender in case of failure
            profilePropagationProtocol.processProfilePropagationRequest(
                     dataBytes, sender);
         break;

         case (int) Protocol.NODE_ADDRESS:
            clientConnectionHandler
                     .processClientRegistration(dataBytes, sender);
         break;

         default:
            System.out.println(moduleName + "Error::Data[1] ->" + dataInfo
                     + " Data length = " + dataBytes.length);
         break;
      }

   }

   /**
    * This is the method which routes different protocol packets to the
    * appropriate protocols in the protocol suite. It does so by investigating
    * the headers in the packets that have been received by any of the receiver
    * threads. The different kind of protocol headers that this method can deal
    * with are contained in <b>cgl.narada.protocol.Protocol</b>.
    * 
    * @param _dataReceived
    *           The data that has been received.
    */

   public void processData(byte[] data, String sender) {
       long totalMemory = Runtime.getRuntime().totalMemory()/1000000;
       long freeMemory = Runtime.getRuntime().freeMemory()/1000000;
       
       //System.out.println(moduleName + "Total = " + totalMemory + "MB  " +
       //                   "Free = " + freeMemory + "MB  " + " Packet Num=" +
       //                   packetNumber);
      //byte[] data = _dataReceived.getData();
      //String sender = _dataReceived.getSender();
      if (data.length == 0) {
         System.out.println("ProtocolHandler:handleData() -> length == 0");
         return;
      }
      int dataInfo = (int) data[0];
      packetNumber++;
      // System.out.println("PH:: Received ->"+new String(data, 0,
      // data.length) +
      // " from " + sender);
      // System.out.println("Received -> protocol Header " + dataInfo +
      // " from " + sender);
      switch (dataInfo) {
         case (int) Protocol.DATA:
            // System.out.println("ProtocolHandler: Data Packets received -> " +
            // sender);
            processDataPackets(data, sender);
         break;

         case (int) Protocol.CONN_SUCC:
            /** CONN_REQ */
            if (ProtocolHandler_Debug)
               System.out
                        .println("ProtocolHandler: Connection successful to -> "
                                 + sender);
         break;

         case (int) Protocol.CONN_SETUP:
            System.out.println("ProtocolHandler: Connection setup request "
                     + "received from -> " + sender);
         break;

         case (int) Protocol.CONN_ALREADY:
            System.out.println("ProtocolHandler: You are already connected to "
                     + sender);
         break;

         case (int) Protocol.NAP_ADDRESS_REQUEST:
            System.out
                     .println("ProtocolHandler: Node Address request received");
            nodeAdditionProtocol.processNodeAddressRequest(data);
         break;

         case (int) Protocol.NAP_ADDRESS_RESPONSE:
            System.out
                     .println("ProtocolHandler: Node Address RESPONSE received"
                              + " from -> " + sender);
            nodeAdditionProtocol.processNodeAddressResponse(data);
         break;

         case (int) Protocol.NAP_CLIENT_ADD:
            System.out
                     .println("ProtocolHandler: Client Address request received");
            clientConnectionHandler.processClientAddressRequest(sender, data);
         break;

         case (int) Protocol.NAP_SETUP_REQUEST:
            System.out.println("ProtocolHandler: Node Setup request received");
            nodeAdditionProtocol.processNodeAdditionRequest(sender, data);
         break;

         case (int) Protocol.NAP_SETUP_RESPONSE:
            System.out.println("ProtocolHandler: Setup response received");
            nodeAdditionProtocol.processNodeAdditionResponse(sender, data);

            // HG: =====
            System.out.println("====== NODE ADDRESS ASSIGNED: "
                     + isNodeAddressAssigned());

            // If the forwarding service is null, then we create one.
            if (fwdService == null && startForwardingService) {
               new Thread() {

                  public void run() {
                     initDiscoveryRequestForwardingService();
                  }
               }.start();
            }
         // ===== !

         break;

         case (int) Protocol.CONNECTION_ADDED:
            System.out.println("ProtocolHandler: Connection Addition received");
            if (connectionPropagation == null) {
               System.out
                        .println("ProtocolHandler:: Connection Propagtion == null");
            }
            connectionPropagation.processConnectionPropagation(data);
         break;

         case (int) Protocol.CONNECTION_LOST:
            System.out.println("ProtocolHandler: Connection Loss received");
            if (connectionPropagation == null) {
               System.out
                        .println("ProtocolHandler:: Connection Propagtion == null");
            }
            connectionPropagation.processConnectionPropagation(data);
         break;

         case (int) Protocol.CONNECTION_BUFFER:
            System.out.println("ProtocolHandler: Connection Buffer received");
            if (connectionPropagation == null) {
               System.out
                        .println("ProtocolHandler:: Connection Propagation == null");
            }

            connectionPropagation.processConnectionBuffer(data);
         break;

         case (int) Protocol.GPP_SETUP_REQUEST:
            System.out.println("ProtocolHandler: Gateway Setup received");
            if (gatewayPropagationProtocol == null)
               System.out
                        .println("ProtocolHandler:: Gateway Propagation == null");
            gatewayPropagationProtocol.handleGatewayCreationRequest(sender,
                     data);
         break;

         case (int) Protocol.GPP_SETUP_RESPONSE:
            System.out
                     .println("ProtocolHandler: Gateway Setup RESPONSE received");
            if (gatewayPropagationProtocol == null)
               System.out
                        .println("ProtocolHandler:: Gateway Propagation == null");
            gatewayPropagationProtocol.handleConnectionSetupResponse(sender,
                     data);
         break;

         case (int) Protocol.PROFILES_EXCHANGE:
            // System.out.println("ProtocolHandler:: Received profiles from " +
            // sender);
            profilePropagationProtocol.processMarshalledProfiles(data);
         break;

         case (int) Protocol.PPP_ADD_PREDICATE:
            System.out.println("ProtocolHandler: PPP ADD received");
            tagvalueProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.PPP_REMOVE_PREDICATE:
            System.out
                     .println("ProtocolHandler: PPP remove predicate received");
            tagvalueProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.PPP_CLIENT_ADD_PREDICATE:
            System.out.print("ProtocolHandler: Client Subscription from "
                     + sender);
            System.out.println(new String(data));
            tagvalueProfilePropagation
                     .processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.JMS_ADD_PROFILE:
            System.out.println("ProtocolHandler: JMS add profile received");
            jmsProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.JMS_REMOVE_PROFILE:
            System.out.println("ProtocolHandler: JMS remove profile received");
            jmsProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.JMS_CLIENT_ADD_PROFILE:
            System.out.println("ProtocolHandler: JMS client add profile "
                     + sender);
            // System.out.println(new String(data));
            jmsProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.JMS_CLIENT_REMOVE_PROFILE:
            System.out.print("ProtocolHandler: JMS client remove profile "
                     + sender);
            System.out.println(new String(data));
            jmsProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.RTP_ADD_PROFILE:
            System.out.println("ProtocolHandler: RTP ADD received");
            rtpProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.RTP_REMOVE_PROFILE:
            System.out.println("ProtocolHandler: RTP remove profile received");
            rtpProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.RTP_CLIENT_ADD_PROFILE:
            System.out.print("ProtocolHandler: RTP client add profile "
                     + sender);
            rtpProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.RTP_CLIENT_REMOVE_PROFILE:
            System.out.print("ProtocolHandler: RTP client remove profile "
                     + sender);
            rtpProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.XPATH_ADD_PROFILE:
            System.out.println("ProtocolHandler: XPath ADD profile received");
            xPathProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.XPATH_REMOVE_PROFILE:
            System.out
                     .println("ProtocolHandler: XPath remove profile received");
            xPathProfilePropagation.handleProfilePropagationRequest(data);
         break;

         case (int) Protocol.XPATH_CLIENT_ADD_PROFILE:
            System.out.print("ProtocolHandler: XPath client add profile "
                     + sender);
            System.out.println(new String(data));
            xPathProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.XPATH_CLIENT_REMOVE_PROFILE:
            System.out.print("ProtocolHandler: XPath client remove profile"
                     + sender);
            System.out.println(new String(data));
            xPathProfilePropagation.processPropagationRequestFromClient(data);
         break;

         case (int) Protocol.CLIENT_EVENT:
            tvEventRoutingProtocol.processEventReceivedFromClient(data);
         break;

         case (int) Protocol.EVENT:
            // System.out.println("ProtocolHandler: EVENT received via ERP
            // relays");
            tvEventRoutingProtocol.processEventReceived(data);
         break;

         case (int) Protocol.NARADA_RTP_CLIENT_EVENT:
            // System.out.println("ProtocolHandler: EVENT received from
            // client");
            rtpEventRoutingProtocol.processEventReceivedFromClient(data);
         break;

         case (int) Protocol.NARADA_RTP_EVENT:
            // System.out.println("ProtocolHandler: EVENT received via ERP
            // relays");
            rtpEventRoutingProtocol.processEventReceived(data);
         break;

         case (int) Protocol.LONG_TOPIC_EVENT:
            this.longTopicGenerator.generate(data);
         break;

         // HG: == Send Discovery Request
         case (int) Protocol.DISCOVERY_REQUEST:
            // =====
            // NOTE: The first byte is (int)Protocol.DISCOVERY_REQUEST
            // Remove the first byte before unmarshalling
            // =====

            byte[] discoBytes = new byte[data.length - 1];
            System.arraycopy(data, 1, discoBytes, 0, discoBytes.length);

            // Also forward this request to other brokers
            processBrokerDiscoveryRequest(discoBytes, true);
         break;

         default:
            System.out.println("Error::Data[0] ->" + dataInfo
                     + " Data length = " + data.length + " Data = " + data
                     + " Packet Num " + packetNumber);
         break;
      }
      data = null;
   }

   /**
    * Responible for garbage collecting a certain link. This would involve
    * grabage collecting the resources associated with the link e.g. sockets,
    * queues and the sender/receiver threads associated with that socket.
    * 
    * @param node
    *           The string using which this node can communicate with a node
    *           that hasn't been assigned a logical address so far
    */
   public void garbageCollectLink(String node) {
      System.out.println("\n ProtocolHandler.garbageCollectLink() needs to"
               + " be implemented");
      // transportHandler.garbageCollectLink(node);
   }

   public void closeCommunicationsTo(String nodeId) {
      transportHandler.closeLinks(nodeId);
   }

   /** Ensures communication links to the specified node are all terminated */
   public void closeCommunicationsTo(Object nodeId) {
      transportHandler.closeLinks(nodeId);
   }

   /**
    * This is invoked by the node which is responsible for issuing the Node
    * setup response. This method performs the following functions
    * <li> First it sends the serialized connection buffer over to the newly
    * added node/unit.
    * <li> Secondly, it adds the newly established connection to the
    * connectivity graph.
    * <li> Initiate a dissemination of the connection information through valid
    * sub systems of the system.
    * 
    * @param createdConnection
    *           The newly established connection
    * @param hop
    *           The gateway over which this information needs to be sent over
    */
   public void propagateConnectionInformation(Connection createdConnection,
            Gateway hop) {
      if (connectionPropagation == null) {
         System.out.println("ProtocolHandler: ConnectionPropagation has not "
                  + "been initialized ->failure in propagation ");
         return;
      }
      /** first send the buffer over */
      connectionPropagation.propagateConnectionBufferToHop(hop);
      /** Then add node to the connectivity graph */
      gatewayInfo.addConnection(createdConnection);

      /** Now disseminate the added connection across the subsystem */
      connectionPropagation.propagateConnectionInformation(createdConnection,
               true);
   }

   /** This method is responsible for initializing the protocol suite. */
   public void initializeProtocolSuite() {
      nodeAdditionProtocol.setProtocolHandler(this);
   }

   public void loadCommunicationsOfType(Properties props, String commType)
            throws TransportException {
      transportHandler.loadCommunicationsOfType(props, commType);
   }

   public String setupLink(Properties props, String linkType)
            throws TransportException {
      return transportHandler.setupLink(props, linkType);
   }

   public void closeLink(String linkId) {
      transportHandler.closeLinks(linkId);
   }

   public String[] enumerateLinks() {
      return transportHandler.enumerateLinks();
   }

   public synchronized void shutdown() {
      if (secureTopicsManager != null) secureTopicsManager.shutdown();

      if (fwdService != null) fwdService.shutdown();

      discoveryRequestTable.clear();
      transportHandler.closeAllLinks();
      transportHandler.shutdown();
   }

   // HG: Security stuff
   /**
    * Handler into the secure topics manager's function to validate the contents
    * of an NBEvent
    * 
    * @param event -
    *           The event to verify
    * @return TRUE if verified, FALSE otherwise
    */
   public boolean doSecurityChecks(NBEvent event) {
      return secureTopicsManager.doSecurityChecks(event);
   }

   public boolean sysnopsisRepresentsSecureTopic(Object synopsis, int type) {
      return secureTopicsManager.sysnopsisRepresentsSecureTopic(synopsis, type);
   }

   public void registerProfileSecurityToken(Object key, byte[] token) {
      secureTopicsManager.registerProfileSecurityToken(key, token);
   }

   public SignedSecurityToken getSecurityTokenForProfileId(Object key) {
      return secureTopicsManager.getSecurityTokenForProfileId(key);
   }

   public boolean verifySignedSecurityToken(SignedSecurityToken sst) {
      return secureTopicsManager.verifySignedSecurityToken(sst);
   }

   public void routeMessageToClient(String reciever, int clientId, byte[] data) {
      clientConnectionHandler.sendMessageToClient(clientId, reciever, data);
   }

   // ----

   /**
    * For communicating with a server node which hasn't been assigned a logical
    * address so far
    * 
    * @param node
    *           The string identifier for the server node.
    * @param data
    *           The byte stream containing the information.
    */
   public void sendTo(String node, byte[] data) {
      try {
         transportHandler.sendData(data, node);
      } catch (TransportException transEx) {
         System.out.println(transEx);
      }
   }

   /**
    * This method send information over a specified hop. The information
    * maintained in the connectivity graph is just the level, and the node
    * address at that level
    */
   public void sendToNode(Gateway nodeAddress, byte[] data) {
      try {
         transportHandler.sendData(data, nodeAddress);
      } catch (TransportException transEx) {
         System.out.println(transEx);
      }
   }

   /**
    * This method is used, when an unassigned node has been assigned logical
    * address by the system
    * 
    * @param node
    *           The string identifier for the server node w/o the logical
    *           address
    * @param _newAddress
    *           The newly assigned address for the node. Henceforth all
    *           communication with the node would be carried out using its
    *           logical address within the system
    */
   public void mapLinkToBrokerNode(String node, Gateway _newAddress,
            ProtocolID brokerConnectionId) {
      transportHandler.mapLinkToBrokerNode(node, _newAddress,
               brokerConnectionId);
   }

   /**
    * This method is used, when an unassigned client node has been assigned
    * logical address by the system
    * 
    * @param node
    *           The string identifier for the client node w/o the logical
    *           address
    * @param _newAddress
    *           The newly assigned address for the node. Henceforth all
    *           communication with the node would be carried out using its
    *           logical address within the system.
    * @param clientIdentifier
    *           The newly assigned identifier for the client
    */
   public void mapLinkToClientNode(String node, Gateway _newAddress,
            Object clientIdentifier) {
      transportHandler.mapLinkToClientNode(node, _newAddress, clientIdentifier);
   }

   public void connectionToClientLost(Object clientIdentifier, String linkId) {
      jmsProfilePropagation.connectionToClientLost(clientIdentifier, linkId);
      rtpProfilePropagation.connectionToClientLost(clientIdentifier);
      // tagvalueProfilePropagation.connectionToClientLost(clientIdentifier);
      profilePropagationProtocol.connectionToClientLost(clientIdentifier,
               linkId);
   }

   public void connectionToBrokerLost(Object brokerConnectionId, String _linkId) {
      if (!gatewayInfo.containsConnection((ProtocolID) brokerConnectionId)) {
         System.out.println(moduleName + "Unknown connectionId "
                  + brokerConnectionId);
         return;
      }

      Connection _lostConnection = gatewayInfo
               .getConnection((ProtocolID) brokerConnectionId);
      System.out.println(moduleName + "Reporting loss of connection "
               + _lostConnection);

      /** Then add node to the connectivity graph */
      gatewayInfo.removeConnection(_lostConnection);

      /** Now disseminate the added connection across the subsystem */
      connectionPropagation.propagateConnectionInformation(_lostConnection,
               false);

      // HG:MGMT
      if (brokerSystemEventListener != null) {
         brokerSystemEventListener.handleEvent(_linkId);
      }
   }

   public void handleData(byte[] data, String sender) {
      processData(data, sender);
      /*
       * long _delay = System.currentTimeMillis() - timeLastProcessed;
       * System.out.println(moduleName + " Inter-packet delay = " + _delay);
       * timeLastProcessed = System.currentTimeMillis();
       */

   }

   public void run() {

   }

   /**
    * Not been implemented so far, but will be soon and would aid as a
    * information gathering tool. The information generated would be used for
    * further optimizations.
    * 
    * @param report
    *           The report being issued by various protocols. All the protocols
    *           have a reference to the protocolHandler.
    */
   public void generateReport(String report) {
      if (ProtocolHandler_Debug) {
         System.out.println("ProtocolHandler: " + report);
      }
   }

   /** This method is response for issuing a node addition request. */
   public byte[] constructNodeAdditionRequest(int[] levels, boolean from) {
      if (nodeAddressAssigned) {
         String errorReport = "\n" + moduleName
                  + "This node has already been assigned the "
                  + "following address: " + thisNodeAddress
                  + ". No further processing will be performed!\n";
         System.out.println(errorReport);
         return null;
      }
      return nodeAdditionProtocol.constructNodeAdditionRequest(levels, true);
   }

   /**
    * Propagates profiles to a newly added node.
    * 
    * @param hop
    * @param level
    */
   public void propagateProfilesAtSpecifiedLevel(Gateway hop, int level) {
      byte[] marshalledProfiles = profilePropagationProtocol
               .marshallProfilesForSpecifiedLevelOnwards(level);
      sendToNode(hop, marshalledProfiles);
   }

   public byte[] constructGatewaySetupRequest(String node, int gatewayLevel) {
      if (gatewayPropagationProtocol == null) {
         System.out.println("This node hasn't been assigned a logical address "
                  + "yet, no way to create gateways between units");
         return null;
      }
      return gatewayPropagationProtocol.constructGatewaySetupRequest(node,
               gatewayLevel);

   }

   public void startMonitoringService() throws TransportException {
      ((TransportHandlerImpl) transportHandler).startMonitoringService();
   }

   public void enableMonitoring() {
      ((TransportHandlerImpl) transportHandler).getMonitoringService()
               .getMeasurementInitiator().enableMonitoring();
   }

   public void disableMonitoring() {
      ((TransportHandlerImpl) transportHandler).getMonitoringService()
               .getMeasurementInitiator().disableMonitoring();
   }

   public void setServiceConfigurationLocation(String configFileLocation) {
      this.configFileLocation = configFileLocation;
   }

   // HG: MGMT:

   /**
    * @param brokerSystemEventListener
    *           The brokerSystemEventListener to set.
    */
   public synchronized final void setBrokerSystemEventListener(
            BrokerSystemEventListener brokerSystemEventListener) {
      this.brokerSystemEventListener = brokerSystemEventListener;
   }

   // HG: =====

   public void initSecureTopicsManager(String keyStoreFileLocation) {
      // Refer Note in constructor
      secureTopicsManager.init(keyStoreFileLocation);
   }

   /**
    * Initializes the Discovery Request forwarding service
    */
   public void initDiscoveryRequestForwardingService() {
      fwdService = new BrokerDiscoveryRequestForwardingService(this,
               configFileLocation);
      ;
   }

   // /**
   // * Initializes the Secure topics' manager communication
   // */
   // public void initSecureTopicsManagerService() {
   // secureTopicsManager.initCommunication(configFileLocation, this);
   // }

   /**
    * Initializes the Discovery Request Table with the specified size
    * 
    * @param size -
    *           MAXIMUM number of entries permitted
    */
   public void initDiscoveryRequestTable(int size) {
      discoveryRequestTable = new HashQueue(size);
   }

   /**
    * Initializes the Broker Discovery Request response policy handler <br>
    * If the specified class cannot be instantiated, the broker defaults to the
    * default policy (i.e. Respond to all requests)
    * 
    * @param handler -
    *           String specifiying the class to load.
    */
   public void setBrokerDiscoveryRequestResponsePolicyHandler(String handler) {
      try {
         requestResponsePolicyHandler = (BrokerDiscoveryRequestResponsePolicy) Class
                  .forName(handler).newInstance();
      } catch (Exception e) {
         System.out.println("*ERROR: Could not load "
                  + "BrokerDiscoveryRequestResponsePolicy:\n" + handler
                  + "\n\t" + e + "\n\t Using default Policy");
         requestResponsePolicyHandler = new DefaultBrokerDiscoveryRequestResponsePolicy();
      }
   }

   public Hashtable getSupportedTransportProtocols() {
      return transportHandler.getSupportedTransportProtocols();
   }

   /**
    * Constructs and send a Broker Discovery Response Message
    * 
    * @param discoRequest -
    *           The bytes that make the BrokerDiscoveryRequest
    * @param mustBeForwarded -
    *           If true, then use the BrokerDiscoveryRequest Forwarding service
    *           to forward the discovery request to other possible brokers in
    *           the network. If FALSE then we do not forward the request.
    */
   public synchronized void processBrokerDiscoveryRequest(byte[] discoBytes,
            boolean mustBeForwarded) {

      BrokerDiscoveryRequest request = BrokerDiscoveryRequest
               .createObjectFromBytes(discoBytes);

      // DEBUGGING purposes only... print all stuff from the request...
      System.out.println(request.toString());

      // Check if this is aduplicate request, If yes, just discard it

      if (discoveryRequestTable.containsKey(request.getRequestUUID())) {
         System.out.println(moduleName + "Duplicate request !");
      } else {
         // Process the request
         // Add for future ref
         discoveryRequestTable.put(request.getRequestUUID(), request);

         // Also fwd this request to other nodes ??
         if ((fwdService != null) && mustBeForwarded) {
            System.out.println(moduleName
                     + "Forwarding request in Broker Network");
            fwdService.forwardRequest(discoBytes);
         }

         // Check against the policy to make sure, We can reply to the
         // client.
         // This also checks for the virtualBroker network requested

         if (requestResponsePolicyHandler.verify(request)) {

            String[] requestedProtocols = request.getTransportProtocols();
            String[] supportedProtocols = transportHandler
                     .getListOfCommunicationProtocolsSupported();

            // O(N^2) comparison here for now.. probably should be
            // changed...
            boolean supported = true;

            for (int i = 0; i < requestedProtocols.length; i++) {
               String protocol = requestedProtocols[i];

               boolean result = false;
               for (int j = 0; j < supportedProtocols.length; j++) {
                  if (supportedProtocols[j].equals(protocol)) {
                     System.out.println(moduleName + "Protocol: [" + protocol
                              + "] supported");
                     result = true;
                     continue;
                  }
               }

               if (!result)
                  System.out.println(moduleName + "Protocol: [" + protocol
                           + "] *NOT* supported");
               supported &= result; // MUST Satisfy ALL MATCH ??

            }

            // No need to send response if this broker instance does not
            // support
            // one or more protocols requested by client.

            if (supported) {
               // Now construct the BrokerDiscoveryResponse
               BrokerAdvertisement ad = new BrokerAdvertisement();

               try {
                  ad.setHostname(InetAddress.getLocalHost().getHostAddress());
               } catch (UnknownHostException e) {
                  System.out.println(moduleName + "ERROR: " + e);
                  ad.setHostname("localhost");
               }
               ad
                        .setSupportedTransportProtocols(getSupportedTransportProtocols());
               ad.setInfo(BrokerNode.aboutThisBroker);

               BrokerDiscoveryResponse discoResponse = new BrokerDiscoveryResponse(
                        request.getRequestUUID());
               discoResponse.setBrokerInfo(ad);
               discoResponse.setNumLinks(enumerateLinks().length);

               byte[] discoResponseBytes = discoResponse.getBytes();

               try {
                  DatagramSocket udpSocket = new DatagramSocket();
                  // InetAddress serverAddress =
                  // InetAddress.getByName(request.getHostname());

                  DatagramPacket outPacket = new DatagramPacket(
                           discoResponseBytes, discoResponseBytes.length,
                           new InetSocketAddress(request.getHostname(), request
                                    .getPort()));

                  System.out.println(moduleName + "Sending "
                           + discoResponseBytes.length + " bytes to <"
                           + request.getHostname() + ":" + request.getPort()
                           + ">");

                  udpSocket.send(outPacket);
                  udpSocket.close();
               } catch (Exception e1) {
                  System.out.println(moduleName
                           + "Error sending BrokerDiscoveryResponse: " + e1);
               }
            }
         } else {
            System.out
                     .println(moduleName + "Policy does not permit response !");
         }
      }

   }

   // ===== !

}
